package com.atguigu.edu.util;


import com.atguigu.edu.common.Constant;

public class SQLUtil {
    public static String getKafkaSourceDDL(String topic, String groupId, String ... dataFormat) {
    
        String format = "json";
        if (dataFormat.length > 0) {
            format = dataFormat[0];
        }
        
        return "with(" +
            " 'connector'='kafka', " +
            " 'properties.bootstrap.servers'='" + Constant.KAFKA_BROKERS + "', " +
            " 'properties.group.id'='" + groupId + "', " +
            " 'topic'='" + topic + "', " +
            " 'scan.startup.mode'='latest-offset', " +
            " 'format'='" + format + "' " +
            ")";
    }
    
    public static String getKafkaSinkDDL(String topic, String ... dataFormat) {
        String format = "json";
        if (dataFormat.length > 0) {
            format = dataFormat[0];
        }
        return "with(" +
            " 'connector'='kafka', " +
            " 'properties.bootstrap.servers'='" + Constant.KAFKA_BROKERS + "', " +
            " 'topic'='" + topic + "', " +
            " 'format'='" + format + "' " +
            ")";
    }
    
    public static String getUpsertKafkaDDL(String topic, String ... dataFormat) {
        String format = "json";
        if (dataFormat.length > 0) {
            format = dataFormat[0];
        }
        return "with(" +
            " 'connector'='upsert-kafka', " +
            " 'properties.bootstrap.servers'='" + Constant.KAFKA_BROKERS + "', " +
            " 'topic'='" + topic + "', " +
            " 'key.format'='" + format + "', " +
            " 'value.format'='" + format + "' " +
            ")";
    }
}
